package com.udf.flink.scala.apitest.table

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.table.api.{EnvironmentSettings, Schema}
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.types.Row
object dimJoin {
  def main(args: Array[String]): Unit = {
    val dim_sql="""CREATE TABLE dim_city (
                  |  city_id varchar,
                  |  city_nm STRING,
                  |  primary key(city_id)  NOT ENFORCED
                  |)
                  |WITH (
                  |    'connector' = 'jdbc',
                  |    'url' = 'jdbc:postgresql://192.168.58.121:5432/dwh',
                  |    'driver' = 'org.postgresql.Driver',
                  |    'table-name' = 'edw.dim_city',
                  |    'username' = 'dcetl',
                  |    'password' = '1q2w3e4r@dcetl'
                  |)""".stripMargin
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val bsSettings = EnvironmentSettings.newInstance().inStreamingMode().build()
    val tableEnv = StreamTableEnvironment.create(env, bsSettings)
    // 获取左侧流
    val orderStream = env
      .socketTextStream("localhost", 9998)
      .map {
        x =>
          val s = x.split(",")
          (s(0), s(1))  // sh,1
      }
    val orderTable = tableEnv.fromDataStream(orderStream,Schema
      .newBuilder()
      .column("_1", "string")
      .column("_2", "string")
      .columnByExpression("p", "PROCTIME()") // .watermark("f1", "`f1` - INTERVAL '5' SECOND")
      .build()
    ).as("oder_nn", "city_id", "p")
    tableEnv.createTemporaryView("orderview", orderTable)
    tableEnv.executeSql(dim_sql)
    val result = tableEnv.sqlQuery(
      """select * from orderview t
        |left join dim_city for SYSTEM_TIME as of t.p as c on t.city_id=c.city_id""".stripMargin)
    tableEnv.toDataStream(result).print()
    env.execute()
  }
}
/*维度表关联
Temporal Join的关键词为FOR SYSTEM_TIME AS OF tb_a.proctime（事件时间为rowtime）

该关联方式相对普通关联有两个好处，一方面该关联方式在实现关联时，使用的是StreamExecLookupJoin，该join可以查询到维度表的变更。普通方式关联jdbc表，在启动的时候加载表数据后，后续变更则无法同步到fllink，所以Temporal Join可以获取到维度表变更；

另一方面普通关联会一直保留关联双侧的数据，数据也就会一直膨胀，直到撑爆内存导致任务失败，Temporal Join则可以定期清理过期数据，在合理的内存配置下即可避免内存溢出；
————————————————
版权声明：本文为CSDN博主「Sword_Zhao」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/zhaochengxuyuan1/article/details/114540312

 */
